package br.unifor.g2cl.hl;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.sf.jgcs.Annotation;
import net.sf.jgcs.ControlSession;
import net.sf.jgcs.DataSession;
import net.sf.jgcs.JGCSException;
import net.sf.jgcs.Message;
import net.sf.jgcs.MessageListener;
import net.sf.jgcs.NotJoinedException;
import net.sf.jgcs.Service;
import net.sf.jgcs.UnsupportedServiceException;
import net.sf.jgcs.membership.Membership;
import net.sf.jgcs.membership.MembershipListener;
import net.sf.jgcs.membership.MembershipSession;
import net.sf.jgcs.utils.ControlSessionUtil;
import br.unifor.g2cl.G2CLMessage;
import br.unifor.g2cl.GroupRequest;
import br.unifor.g2cl.IMarshalDataSession;
import br.unifor.g2cl.SuspectedException;
import br.unifor.g2cl.TimeoutException;
import br.unifor.g2cl.Util;
import br.unifor.g2cl.hl.listener.MessageDispatcherListener;


/**
 * Provides a way to send synchronous messages to the group with request-response correlation. Synchronous messages should be sent following the policies defined on {@link GroupRequest}.
 * <br/><br/>
 * This service is part of the High-level group.
 * 
 * @author Leandro Sales (leandro.shp@gmail.com)
 */
public class MessageDispatcher implements MessageListener, MembershipListener{

	private static Logger logger = Logger.getLogger(MessageDispatcher.class.getName());
	
	/* Internal message headers */
	
	private static final byte[] REQUEST = new byte[] { (byte)-121 };
	private static final byte[] RESPONSE = new byte[] { (byte)-122 };
	private static final byte[] REGULAR_MESSAGE = new byte[] { (byte)-123 };

	private List<SocketAddress> currentView = new ArrayList<SocketAddress>();
	private ConcurrentHashMap<Integer, RspList> requests;
	private AtomicInteger nextRequestId = new AtomicInteger(0);

	/** Receive its own broadcast messages to the group. Default is true. */
	private boolean local = true;

	// jgcs related objects
	protected IMarshalDataSession marshalDataSession;
	protected MembershipSession membershipSession;
	protected Service service;
	private MembershipListener membershipListener;
	private MessageDispatcherListener messageDispatcherListener;
	private MessageListener messageListener;

	private Marshaller rspMarshaller = null;


	public MessageDispatcher(	IMarshalDataSession marshalDataSession,
								MembershipSession membershipSession, 
								Service service) throws JGCSException {

		this.marshalDataSession = marshalDataSession;
		this.membershipSession = membershipSession;
		this.service = service;
		this.requests = new ConcurrentHashMap<Integer, RspList>();
		
		this.marshalDataSession.setMessageListener(this);
		
		ControlSessionUtil.redefineMembershipListener(this.membershipSession, this);

		if (membershipSession.isJoined()) {
			setMembers(this.membershipSession.getMembership().getMembershipList());
		}
	}

	public void setMessageListener(MessageListener listener) {
		this.messageListener = listener;
	}
	
	public List<SocketAddress> getMembers() {
		synchronized (currentView) {
			return new ArrayList<SocketAddress>(currentView);
		}
	}
	
	private void setMembers(List<SocketAddress> list) {
		synchronized (currentView) {
			currentView.clear();
			currentView.addAll(list);
		}
	}
	
	public Marshaller getRspMarshaller() {
		return rspMarshaller;
	}

	public void setRspMarshaller(Marshaller rspMarshaller) {
		this.rspMarshaller = rspMarshaller;
	}
	

	public void setMessageDispatcherListener(
			MessageDispatcherListener messageDispatcherListener) {
		this.messageDispatcherListener = messageDispatcherListener;
	}

	
	/**
	 * To set your MembershipListener since this object will set itself
	 * as the listener.
	 * 
	 * @param membershipListener
	 */
	public void setMembershipListener(MembershipListener membershipListener) {
		this.membershipListener = membershipListener; 
	}
	
	public RspList castMessage(	final Vector<SocketAddress> dests, Message msg, 
			int mode, long timeout) throws UnsupportedServiceException, IOException {
		
		return castMessage(dests, msg, mode, timeout,false,null);
	}
	
	protected List<SocketAddress> getDestinations(	final Vector<SocketAddress> dests){
		
		List<SocketAddress> sendToAddress = new ArrayList<SocketAddress>();
		List<SocketAddress> view = getMembers();
		for (SocketAddress address : view) {
			if (local || !address.equals(membershipSession.getLocalAddress())) {
				sendToAddress.add(address);
			}
		}
		return sendToAddress;
	}
	
	public RspList castMessage(	final Vector<SocketAddress> dests, 
								Message msg, 
								int mode, 
								long timeout, 
								boolean useAnycasting,
								RspFilter filter) throws UnsupportedServiceException, IOException {
		
		G2CLMessage message = (G2CLMessage) msg;
		boolean sendToAllMembers = dests == null;
			
		List<SocketAddress> sendToAddress = sendToAllMembers? getDestinations(dests) : dests;
		
		// getting all members
		int requestID = getNextRequestId();
		RspList request = new RspList(requestID, mode, filter);

		request.addRsps(sendToAddress);

		requests.put(requestID, request);

		// compositing header
		byte[] dispatcherHeader = composeRequestHeader(requestID);
		message.addHeader(dispatcherHeader);

		if (useAnycasting || !sendToAllMembers) {
			for (SocketAddress otherMember : sendToAddress) {
				marshalDataSession.send(message, service, null, otherMember,
						(Annotation[]) null);
			}
		} else {
			marshalDataSession.multicast(message, service, null,
					(Annotation[]) null);
		}

		if (request.size() > 0) {
			request.waitResponses(timeout);
		}
		requests.remove(requestID);

		return request;
	}

	/**
	 * 	 * @throws IOException
	 * @throws UnsupportedServiceException
	 */
	public Object sendMessage(Message msg, SocketAddress dest, int mode,
			long timeout) throws TimeoutException, SuspectedException,
			UnsupportedServiceException, IOException {
		
		G2CLMessage message = (G2CLMessage) msg;

		if (dest == null) {
			// FIXME
			throw new RuntimeException(
					"dest must be not empty !");
		}

		int requestID = getNextRequestId();
		RspList request = new RspList(requestID, mode);
		request.addRsp(dest);

		requests.put(requestID, request);

		byte[] dispatcherHeader = composeRequestHeader(requestID);
		message.addHeader(dispatcherHeader);

		marshalDataSession.send(message, service, null, dest, (Annotation[])null);
		
		request.waitResponses(timeout);

		// since this method is used for a single member, it should have only one response...
		Rsp response = (Rsp) request.values().iterator().next();

		if (!response.wasReceived() && !response.wasSuspected()) {
			throw new TimeoutException();
		} else if (!response.wasReceived() && response.wasSuspected()) {
			throw new SuspectedException();
		}

		return response.getValue();
	}

	
	
	public Object onMessage(Message msg) {

		G2CLMessage message = (G2CLMessage) msg;
		SocketAddress sender = message.getSenderAddress();

		SocketAddress localAddr = membershipSession.getLocalAddress();
		
		// ignore own messages?
		if (sender.equals(localAddr) && !local) {
			return null;
		}

		try {
			byte[] dispactherHeader = message.removeHeader();

			if (dispactherHeader[0] == REQUEST[0]) {
				int requestId = Util.byteArrayToInt(dispactherHeader, 1);
				handleRequest(requestId, message, localAddr);

			} else if (dispactherHeader[0] == RESPONSE[0]) {
				int requestId = Util.byteArrayToInt(dispactherHeader, 1);
				handleResponse(requestId, message);
			} else if (dispactherHeader[0] == REGULAR_MESSAGE[0]) {
				if (messageListener != null) {
					messageListener.onMessage(msg);
				}
			} else {
				// XXX
				if (logger.isLoggable(Level.WARNING)) {
					logger.log(Level.WARNING, "Received unknown message. " +
					"It is not a MessageDispatcher message");
				}
			}
		} catch (Exception e) {
			logger.log(Level.SEVERE, "Error receiving message", e);
		}
		return null;
	}
	
	private void handleResponse(int requestId, G2CLMessage message){
		RspList request = requests.get(requestId);
		// If it isn't end yet...
		if (request != null) {
			Object returnVal = null;
			try {
				byte[] m = message.getPayload();
				if (rspMarshaller != null) {
					returnVal = rspMarshaller.getObjectFromByte(m, 0, m.length); 
				} else {
					returnVal = Util.getObjectFromByte(m, 0, m.length);
				} 
				
			} catch (Exception e) {
				logger.log(Level.SEVERE, "Error while deserializating object", e);
			}
			request.receiveResponse(message.getSenderAddress(), returnVal);
		}
	}

	
	private void handleRequest(	int requestId, 
								G2CLMessage message, 
								SocketAddress localAddress) throws UnsupportedServiceException, IOException{
		
		G2CLMessage resp = (G2CLMessage) marshalDataSession.createMessage();
		if (messageDispatcherListener != null) {
			byte[] r = null;
			try {
				if (rspMarshaller != null) {
					r = rspMarshaller.getArrayFromObject(messageDispatcherListener.handle(message)); 
				} else {
					r = Util.getArrayFromObject(messageDispatcherListener.handle(message));
				} 
			} catch (Exception e) {
				logger.log(Level.SEVERE, "Error while serializating object", e);
			}
			resp.setPayload(r);
		}
		else {
			resp.setPayload(null); // to not fuck up addHeader
			logger.log(Level.FINE, "Handle for message dispatcher not set. Are you sure??");
		}
		
		resp.setSenderAddress(localAddress);
		
		byte[] responseHeader = composeResponseHeader(requestId);
		resp.addHeader(responseHeader);
		marshalDataSession.send(resp, service, null, message.getSenderAddress(), (Annotation[])null);
	}

	
	private byte[] composeRequestHeader(int requestID){
		// compositing header
		byte[] dispatcherHeader = new byte[5];
		// setting message type as request
		dispatcherHeader[0] = REQUEST[0];
		// setting request id
		Util.intToByteArray(dispatcherHeader, 1, requestID);
		return dispatcherHeader;
	}
	
	private byte[] composeResponseHeader(int requestId) {
		byte[] responseHeader = new byte[5];
		responseHeader[0] = RESPONSE[0];
		Util.intToByteArray(responseHeader, 1, requestId);
		return responseHeader;
	}

	public void stop() {
		messageDispatcherListener = null;
		requests.clear();
	}

	public ControlSession getControlSession() {
		return (ControlSession) membershipSession;
	}

	public DataSession getDataSesson() {
		return marshalDataSession;
	}
	
	public Service getService() {
		return service;
	}
	
	public int getNextRequestId() {
		return nextRequestId.incrementAndGet();
	}

	/** Determines whether local responses should be processed. */
	public boolean isLocal() {
		return local;
	}

	/** Sets whether local responses should be processed. Default is true */
	public void setLocal(boolean local) {
		this.local = local;
	}

	/* ********************* MEMBERSHIPLISTENER ************************ */
	
	public void onExcluded() {
		if (membershipListener != null) {
			membershipListener.onExcluded();
		}
	}

	public void onMembershipChange() {
		try {
			Membership membership = membershipSession.getMembership();
			setMembers(membership.getMembershipList());
			List<SocketAddress> failedMembers = membership.getFailedMembers();
			List<SocketAddress> leavedMembers = membership.getLeavedMembers();

			Collection<RspList> rsps = requests.values();
			for (RspList rspList : rsps) {
				if(failedMembers != null){
					for (SocketAddress failedMember : failedMembers) {
						Rsp rsp = rspList.get(failedMember);
						if (rsp != null) {
							rspList.addSuspect(failedMember);
						}
					}					
				}
				if(leavedMembers != null){
					for (SocketAddress leavedMember : leavedMembers) {
						Rsp rsp = rspList.get(leavedMember);
						if (rsp != null) {
							rspList.addSuspect(leavedMember);
						}
					}					
				}
			}
		} catch (NotJoinedException e) {
			logger.log(Level.SEVERE, "Not joined!", e);
		}
		
		if (membershipListener != null) {
			membershipListener.onMembershipChange();
		}		
	}

	public RspList castMessage(Vector<SocketAddress> dests, Message msg,
			int mode, long timeout, boolean useAnycasting)
			throws UnsupportedServiceException, IOException {
		
		return castMessage(dests, msg, mode, timeout, useAnycasting, null);
	}
	
	
	//////////
	
	
	public void multicast(Message msg, Service service, Object cookie, Annotation... annotation) 
	throws IOException, UnsupportedServiceException {
		G2CLMessage m = (G2CLMessage) msg;
		m.addHeader(REGULAR_MESSAGE);
		marshalDataSession.multicast(m, service, cookie, annotation);
	}


	public void send(Message msg, Service service, Object cookie, SocketAddress destination, 
			Annotation... annotation) throws IOException, UnsupportedServiceException {
		G2CLMessage m = (G2CLMessage) msg;
		m.addHeader(REGULAR_MESSAGE);
		marshalDataSession.send(m, service, cookie, destination, annotation);

	}
}